GitHub：darkstar1227/learning-sre-for-ai-era

結論先說：把①到⑦談的每個概念變成可以真的跑起來的東西，第一步不是急著接 SDK，而是先想清楚 trace 邊界要對準工程責任、dataset 要能版本化——這篇（中）就是把這兩件事變成可執行的程式碼。

承接上文：上一篇（上）談了 LangSmith 在整條觀測鏈裡的定位、五個容易搞混的 ID（request_id／task_id／trace_id／evaluation_run_id／dataset_example_id）、project／metadata／tags 的分工，以及 dataset 要把「已知失敗」寫成可重跑的題目、evaluation 要優先用 deterministic evaluator 而非平均分數決定放行。這一篇開始動手：把前面七段的原則接上 LangSmith SDK，讀者可在自己的隔離環境重現。

⑧ 今日 DIY：把 SRE Lab 接到 LangSmith

以下由讀者在自己的隔離環境執行。本文不建立 DIY/、不安裝套件、不啟動服務，也不會宣稱結果已驗證。

這十六個 Step 不是十六件互不相干的小任務，而是把前面①到⑦談的每個論點，一步步變成可以真的跑起來的東西。讀懂這個對應關係，比逐行照抄程式碼更重要：

Step 1-2   驗證 ① 的「LangSmith 該放在哪」——先確認 trace 真的能送出去，
           再確認 workflow 的邊界是照工程責任畫的，不是隨便挑幾個函式
Step 3-5   驗證 ② 的「五個 ID 如何串起 investigation」——建立
           trace data contract，決定哪些欄位能進 trace、哪些不能
Step 6-7   驗證 ③ 的「project/metadata/tags 分工」，並用兩筆
           request 具體看出「HTTP 200 之外還有 task_status 的差異」
Step 8-9   驗證 ④⑤ 的「dataset 不是備份、evaluator 從 deterministic
           開始」——把抽象的 JSON 範例變成真的可執行的 dataset
Step 10    驗證 ⑤ 談到的 LLM-as-judge 防線，包含 Claude 案例點出的
           「evaluator 本身也可能失準」該怎麼收斂成具體規則
Step 11-12 驗證 ⑥ 的 feedback triage 流程，把上面 triage 走查的
           時間軸變成一份可執行的 investigation runbook
Step 13-14 驗證 ⑦ 的 release gate 與 learning loop，把決策接回
           既有的 metrics/logs/trace 系統
Step 15-16 收斂：列出容易犯的錯，並整理一份可以拿去給人審查的
           evidence 包

如果讀者只想抓重點、不打算真的動手跑，讀完每個 Step 的說明文字仍然能知道「這一步在驗證前面哪個論點」；真正想動手驗證的讀者，才需要逐行照著程式碼跑一次。這兩種讀法都是本文預期的使用方式。

Step 1：建立獨立 project 與最小設定

在 Day 0 的環境確認已有 LangSmith account 與 API key。將 key 放在 .env 或安全的 secret store，不要 commit。

uv add langsmith
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="..."
export LANGSMITH_PROJECT="sre-lab-dev"

再依 Day 2 的方式啟動 FastAPI service。預期在指定 project 看見新的 root run；若沒有 trace，先檢查環境變數、SDK 初始化與 request 是否真的走過受追蹤的函式。

這一步在驗證什麼：①段落主張「project 是環境與工作流的分隔邊界」。這一步的目的不是把 tracing 打開就結束，而是確認 LANGSMITH_PROJECT=sre-lab-dev 這個環境變數真的能把 trace 導到正確的地方——如果讀者接下來又建立一個 sre-lab-staging 卻忘了改這個變數，兩個環境的 trace 會混在同一個 project 裡，之後很難靠肉眼分辨。

常見的坑：最常見的失敗不是 SDK 報錯，而是「什麼都沒發生」——服務正常回應，UI 上卻看不到新的 run。原因通常是三選一：環境變數在啟動服務的那個 shell session 沒有生效（尤其是用 .env 檔卻忘了載入）、API key 打錯或過期、或是 request 走的程式碼路徑根本沒有被 @traceable 包住。LangSmith 官方 SDK 預設會靜默吞掉追蹤失敗（避免因為觀測系統掛掉連帶讓業務請求失敗），所以「trace 沒出現」不會讓 API 回應報錯，這點容易讓人誤以為系統正常運作。

Step 2：讓 workflow run 有清楚邊界

以 SDK 的 traceable 包住 workflow；依你的 provider 與 retrieval 實作補上內部函式。

from langsmith import traceable

@traceable(name="answer-workflow", run_type="chain")
def answer_workflow(question: str, *, request_id: str, task_id: str) -> dict:
    documents = retrieve(question)
    answer = call_model(question, documents)
    result = validate(answer, documents)
    return {
        "request_id": request_id,
        "task_id": task_id,
        "answer": answer,
        "citations": result.citations,
        "quality_status": result.status,
    }

在 root run 或子 run 記錄 request_id、task_id、deployment、prompt 與 index version。只存你願意交給追蹤系統保存的資料；原始敏感內容改為 hash、長度、版本或遮罩後的分類值。

這一步在驗證什麼：②段落強調「function boundary 要對準工程責任，不是隨便切」。這一步的重點不在於 @traceable 這個 decorator 本身多好用，而在於讀者要先想清楚「retrieve、model、validate 各自對應哪個團隊、哪一種失敗」，再決定要不要把它包成獨立的 run。如果把整個 workflow 包在一個函式裡、只掛一層 @traceable，Step 7 之後想要「只看 retrieval 是不是拿到空結果」就無從查起——trace 只會有一個扁平的 root run，看不出內部發生了什麼。

預期會看到什麼：跑完這一步，LangSmith UI 上應該能看到一個 root run（answer-workflow），底下至少有一層子 run。如果子 run 沒有出現，最常見的原因是內部函式呼叫時繞過了被 decorate 的版本（例如在同一個模組裡直接呼叫沒有加上 @traceable 的私有函式），或是子函式是用 async 定義卻沒有用對應的非同步呼叫方式，導致追蹤 context 沒有正確傳遞下去。

Step 3：建立 regression dataset 與 rule-based evaluator

將上方 JSON 寫成 dataset example。使用 LangSmith SDK 建立 dataset 後，以 baseline 和 candidate 各跑一次 evaluation。保留 dataset、evaluator、candidate 的版本與執行時間，否則日後無法重現 release decision。

release_decision = (
    "blocked"
    if evaluation.required_source_failures > 0 or evaluation.safety_failures > 0
    else "human_review_required"
    if evaluation.quality_regressions > 0
    else "allowed"
)

這是示意規則。門檻、owner 與例外要由產品風險、SLO policy 與安全流程共同決定。

這一步在驗證什麼：這段 Python 只有五行，卻是整條迴圈第一次把④⑤兩段的抽象主張變成真的可以被程式呼叫的邏輯。讀者在這一步不需要真的跑起完整的 LangSmith 呼叫，重點是先在腦中（或真的寫一小段測試）把這三個分支想清楚：required_source_failures 和 safety_failures 任何一項大於零就直接 blocked，不會被 quality_regressions 的高低沖淡——這正是 Step 9 之後會具體展開的「平均分數不能替風險分類做決定」的最小示例。

Step 4：把一次 failure trace 轉成 regression case

選 Day 30 的 provider、DB 或 tool 實驗之一，或一筆已完成遮罩的真實 incident：

1. 由 request_id 找到 root trace 和相關 child run。
2. 保留 failure class、版本、outcome 與必要 evidence，移除敏感內容。
3. 寫成 dataset example，包含 expected behavior 與 forbidden behavior。
4. 對 baseline、candidate 跑 evaluator。
5. 產生 allowed、human_review_required 或 blocked 的 release decision report，附原因與 owner。

這一步在驗證什麼：這是全篇 DIY 裡唯一一個「不從零開始」的 Step——它要求讀者回收 Day 30 已經做過的 chaos experiment，而不是無中生有想一個新案例。這樣安排的用意，是驗證⑦段落強調的「learning loop 要接上既有證據鏈，不要另起一套」：如果讀者發現 Day 30 產生的 evidence（provider timeout、DB 不可用、tool 誤用）沒辦法直接拿來寫成 dataset case，通常代表當時的實驗紀錄本身就少了「expected behavior」這個欄位，值得回頭把 Day 30 的紀錄格式補齊，而不是在這裡重新編一個案例。

容易踩的坑：第三步的「forbidden behavior」常常被忽略，團隊只寫「應該回答什麼」，卻沒寫「絕對不該做什麼」（例如未經授權呼叫寫入型 tool）。deterministic evaluator 只能檢查程式碼裡明確寫出的條件，一個沒被寫進 forbidden behavior 的危險行為，不會因為「大家都知道不該這樣做」就自動被攔下來。

Step 5：先畫出 trace data contract，再接 SDK

不要一開始就把每個 Python 物件、prompt 或文件塞進 trace。先決定 investigation 時真正要回答的問題。

這個 request 由哪個 deployment 處理？
用了哪個 prompt、模型、retrieval index 與 tool policy？
結果屬於 completed、degraded、rejected 還是 failed？
若品質出問題，應回到哪個 child run？

下表是 Lab 可以採用的最小 contract。欄位名稱可以依服務調整；重要的是定義清楚、在 logs、OTel 與 AI trace 間能對得起來。

request_id、task_id 和 deployment 版本是 investigation join key，不是 metrics label。Prometheus 把每個 request 變成 label 只會產生高基數 time series；把細節留在 trace 與 log，metrics 只保留有限的 status、route、model route 或 outcome 類別。

先在服務內建立資料結構。它不是 LangSmith 專用型別，日後換到 Langfuse、OTel backend 或資料倉儲都能沿用。

from dataclasses import dataclass


@dataclass(frozen=True)
class WorkflowContext:
    request_id: str
    task_id: str
    workflow_name: str
    workflow_version: str
    service_version: str
    deployment_id: str
    prompt_version: str
    policy_version: str
    retrieval_index_version: str
    model_route: str

接著把它轉成只含低敏感度值的 mapping。這一步是遮罩邊界，不是裝飾。

def trace_metadata(context: WorkflowContext) -> dict[str, str]:
    return {
        "request_id": context.request_id,
        "task_id": context.task_id,
        "workflow_name": context.workflow_name,
        "workflow_version": context.workflow_version,
        "service_version": context.service_version,
        "deployment_id": context.deployment_id,
        "prompt_version": context.prompt_version,
        "policy_version": context.policy_version,
        "retrieval_index_version": context.retrieval_index_version,
        "model_route": context.model_route,
    }

如果某個欄位可能含人名、文件標題、帳號或 query 原文，先不要納入。需要關聯時，使用服務內部可解析的 surrogate ID，並確認它的 retention 與存取權限。hash 也不是自動安全：小字典中的問題、email 或員工編號仍可能被反推。

一個常見的過度收集範例，值得對照著看：

# 過度收集：把整個 request 和 response 塞進 metadata
trace_metadata_bad = {
    "raw_question": question,              # 可能含個資或內部機密
    "raw_documents": [doc.text for doc in documents],  # 完整文件正文
    "raw_answer": answer["answer"],        # 完整回覆內容
    "user_email": user.email,              # 直接違反不放個資的原則
}

# 對照 Step 5 的 trace_metadata()：只留下版本與分類值，
# 原文與個資留在既有的、有存取控制的服務資料庫裡，
# trace 只保留「查得到對應紀錄」所需要的 surrogate ID

左邊這種寫法在 Lab 階段很容易發生，因為「反正先全部記下來，之後要查什麼都有」聽起來很誘人。但②段落已經說明過，這些欄位一旦進了 trace，就要承擔 trace 系統自己的 retention 與存取控制標準，而不是服務資料庫的標準——兩者通常不一樣，trace 系統的存取範圍往往比業務資料庫更寬鬆，因為它的設計目的是給工程團隊除錯用，不是給業務系統做權限隔離。

這一步在驗證什麼：WorkflowContext 這個 dataclass 是整篇 DIY 裡最容易被低估的一段程式碼，因為它看起來只是把幾個字串包在一起。它真正驗證的，是②段落主張的「先決定要回答的問題，再決定要記什麼欄位」——frozen=True 這個設定不是隨手加的裝飾，而是刻意讓這個 context 物件建立後不能被中途竄改，逼著開發者在服務邊界就把所有版本資訊準備齊全，而不是讓 retrieval、model、validator 各自在執行過程中偷偷修改共用的版本欄位。

跑的時候預期看到什麼：trace_metadata() 執行後應該得到一個純字串的 dict，讀者可以直接拿它跟 Step 3 提到的 metadata JSON 範例比對欄位是否一致。如果讀者把這個 dict 直接丟給 LangSmith 的 metadata 參數，之後在 UI 上應該能依 prompt_version 或 deployment_id 篩選出對應的 run——這正是驗證③段落「metadata 適合查詢，不是垃圾抽屜」的方式：篩不出來，通常代表某個欄位在傳遞過程中被漏掉或型別不對（例如把版本號存成整數而非字串，導致篩選條件比對失敗）。

Step 6：讓 root trace 和 child run 反映真實 workflow

LangSmith 官方 Python SDK 可用 @traceable 為函式建立 run。先把 function boundary 對準工程責任，而不是對準「我剛好有幾個函式」。

answer_workflow       使用者工作的一次 root trace
├─ retrieve           查詢知識來源與取得候選文件
├─ build_prompt       選擇 prompt 和政策版本
├─ generate           呼叫模型或 fallback route
├─ execute_tool       經政策允許後執行外部動作
└─ validate_response  檢查引用、品質、安全與 task outcome

以下範例刻意用假的 dependency。它讓讀者在隔離環境先驗證 trace 結構，不會真的向模型、資料庫或 tool provider 發出請求。

from dataclasses import dataclass

from langsmith import traceable


@dataclass(frozen=True)
class RetrievedDocument:
    document_id: str
    title: str
    text: str


@traceable(name="retrieve-policy", run_type="retriever")
def retrieve_policy(question: str, index_version: str) -> list[RetrievedDocument]:
    if "遠端" not in question:
        return []
    return [
        RetrievedDocument(
            document_id="remote-work-policy-v3",
            title="Remote Work Policy",
            text="遠端工作須經主管核准。",
        )
    ]


@traceable(name="build-answer-prompt", run_type="prompt")
def build_answer_prompt(question: str, documents: list[RetrievedDocument]) -> dict:
    sources = [document.document_id for document in documents]
    return {"question": question, "source_ids": sources}


@traceable(name="generate-answer", run_type="llm")
def generate_answer(prompt: dict) -> dict:
    if not prompt["source_ids"]:
        return {"answer": "資料不足，無法確認。", "citations": []}
    return {
        "answer": "需要主管核准。",
        "citations": ["remote-work-policy-v3"],
    }

validator 也要獨立成 run。否則 trace 會顯示模型產出，卻無法看出服務是如何決定回覆能不能交給使用者。

@traceable(name="validate-answer", run_type="chain")
def validate_answer(answer: dict, documents: list[RetrievedDocument]) -> dict:
    available = {document.document_id for document in documents}
    citations = set(answer["citations"])

    if not documents:
        return {
            "quality_status": "insufficient_context",
            "safety_status": "pass",
            "task_status": "degraded",
            "citation_valid": True,
        }

    citation_valid = citations.issubset(available)
    return {
        "quality_status": "pass" if citation_valid else "citation_invalid",
        "safety_status": "pass",
        "task_status": "completed" if citation_valid else "failed",
        "citation_valid": citation_valid,
    }

root workflow 回傳的不是任意文字，而是能被 log、trace、evaluation 和 API response 共用的 outcome contract。

@traceable(name="answer-workflow", run_type="chain")
def answer_workflow(question: str, context: WorkflowContext) -> dict:
    documents = retrieve_policy(question, context.retrieval_index_version)
    prompt = build_answer_prompt(question, documents)
    answer = generate_answer(prompt)
    validation = validate_answer(answer, documents)

    return {
        "request_id": context.request_id,
        "task_id": context.task_id,
        "answer": answer["answer"],
        "citations": answer["citations"],
        "document_count": len(documents),
        **validation,
    }

若你的服務使用 LangChain Runnable，官方文件也支援在 invoke() 時傳入 tags 與 metadata，並由 child run 繼承。這對 candidate、canary、incident-123 這類短期篩選有用，但不要將 tag 當成永久資料庫 schema。

result = chain.invoke(
    {"input": "遠端工作是否需要主管核准？"},
    {
        "tags": ["staging", "candidate"],
        "metadata": {
            "prompt_version": "policy-v2",
            "retrieval_index_version": "policy-v3",
        },
    },
)

這段僅適用於已採用 LangChain Runnable 的服務。純 Python workflow 不必為了加 metadata 強行引入 LangChain；先用 SDK 的 trace 結構與既有 OTel context 保持邊界清楚。

這一步整體在驗證什麼：retrieve_policy、build_answer_prompt、generate_answer、validate_answer 四個函式合起來，是②段落「function boundary 要對準工程責任」最具體的落地版本。特別值得注意的是 validate_answer 被獨立成一個函式而不是塞在 generate_answer 尾巼——這對應到⑦段落 Step 12 investigation runbook 裡的第六條判準：「validator 是否放過不應放行的回覆」。如果 validator 邏輯藏在模型呼叫函式裡，事後 trace 只會顯示「模型輸出了什麼」，看不出服務是依據什麼規則判定這次輸出可不可以交給使用者，Step 12 的判準就沒有對應的 child run 可以查。

跑起來預期看到什麼：用「遠端工作是否需要主管核准」這個問題呼叫 answer_workflow，validate_answer 應該回傳 quality_status: pass、task_status: completed；換成 dataset 裡沒有涵蓋的問題（例如問供應商折扣），retrieve_policy 會回傳空列表，validate_answer 應該改回傳 quality_status: insufficient_context、task_status: degraded。這兩種結果的差異，正是下一步 Step 7 用斷言驗證的重點。

容易踩的坑：這裡的範例刻意用了假的 retrieve_policy——只用字串包含判斷「遠端」兩個字就決定要不要回傳文件。讀者若照抄去接真實的向量檢索，第一次很可能會遇到「retrieval 回傳了看似相關但其實不對的文件」，而不是回傳空結果——這正是①段落提到的「semantic failure」，validate_answer 目前的實作只檢查 citation 是否存在於已知文件集合裡，並不檢查文件內容是否真的支持答案，這是本文刻意保留給讀者的延伸空間，不是本文示意的 validator 邏輯已經完備。

Step 7：用兩筆 request 驗證 trace 結構，而不是只看 UI 漂亮

在隔離 Lab 準備一筆正常案例和一筆應該降級的案例。兩者都應回 HTTP 200 或其他成功 transport status 嗎？取決於你的 API contract；本文不替服務做決定。重要的是 outcome 必須不同、可被使用者與 SLI 定義辨識。

normal = answer_workflow(
    "公司遠端工作是否需要主管核准？",
    WorkflowContext(
        request_id="req-lab-001",
        task_id="task-policy-001",
        workflow_name="policy-qa",
        workflow_version="2026-09-22",
        service_version="answer-api-0.9.0",
        deployment_id="lab-local",
        prompt_version="policy-v2",
        policy_version="policy-guard-v1",
        retrieval_index_version="policy-v3",
        model_route="fake-model",
    ),
)

unknown = answer_workflow(
    "供應商下季的折扣是多少？",
    WorkflowContext(
        request_id="req-lab-002",
        task_id="task-policy-002",
        workflow_name="policy-qa",
        workflow_version="2026-09-22",
        service_version="answer-api-0.9.0",
        deployment_id="lab-local",
        prompt_version="policy-v2",
        policy_version="policy-guard-v1",
        retrieval_index_version="policy-v3",
        model_route="fake-model",
    ),
)

assert normal["task_status"] == "completed"
assert unknown["task_status"] == "degraded"
assert unknown["quality_status"] == "insufficient_context"

預期觀測不是「畫面有一條 trace」而已：

req-lab-001
  root: answer-workflow
  child runs: retrieve-policy → build-answer-prompt → generate-answer → validate-answer
  outcome: completed / pass / pass
  citations: remote-work-policy-v3

req-lab-002
  root: answer-workflow
  child runs: 同樣完整，retrieve-policy 的結果為空
  outcome: degraded / insufficient_context / pass
  citations: []

第二條 trace 的價值在於證明「沒有資料」不是模型失敗後偷偷編答案。若 UI 只顯示 root run，先確認 decorator 是否在同一個 request path 執行；若 child run 缺失，不要用手工記錄幾個文字欄位假裝 workflow 已被觀測。

Step 8：建立可版本化的 regression dataset

dataset 的單位是「可重跑的代表案例」，不是原始 production trace 的備份。先為每一筆案例寫出來源、遮罩方式、期望行為與風險層級。

from langsmith import Client


client = Client()
dataset = client.create_dataset(
    dataset_name="policy-task-dataset-v1",
    description="Sanitized policy Q&A regression cases for the SRE Lab.",
)

client.create_examples(
    dataset_id=dataset.id,
    examples=[
        {
            "inputs": {"question": "公司遠端工作是否需要主管核准？"},
            "outputs": {
                "required_sources": ["remote-work-policy-v3"],
                "required_fact": "需要主管核准",
                "expected_task_status": "completed",
            },
            "metadata": {
                "category": "policy_qa",
                "risk_level": "medium",
                "source": "sanitized-lab-case",
            },
        },
        {
            "inputs": {"question": "供應商下季的折扣是多少？"},
            "outputs": {
                "required_sources": [],
                "required_fact": "",
                "expected_task_status": "degraded",
            },
            "metadata": {
                "category": "insufficient_context",
                "risk_level": "medium",
                "source": "sanitized-lab-case",
            },
        },
    ],
)

這個例子不含真實公司政策，也不應假裝它已經代表 production traffic。日後要從 incident 或 feedback 建立 case 時，先完成下列審查：

1. 原始 trace 是否含個資、內部內容、密碼、access token 或合約資訊？
2. 若有，遮罩後還保留了能測試 failure 的最低必要事實嗎？
3. 是單一使用者特例，還是可能重複發生的 workflow failure？
4. expected output 是否由能負責的人確認，而非只採納使用者的倒讚？
5. 將案例加入 dataset 後，誰可以查看、編輯、刪除與標記版本？

dataset 會演進。新增或修改案例前，將修改理由放在 PR、change manifest 或 dataset metadata；標記一個固定版本供 release candidate 使用。LangSmith 文件提供 dataset version tag，讓團隊以語意名稱固定特定時間點，而不是讓 CI 每次默默拿到更新中的 dataset。

from datetime import datetime, timezone

client.update_dataset_tag(
    dataset_name="policy-task-dataset-v1",
    as_of=datetime.now(timezone.utc),
    tag="release-candidate-2026-09-22",
)

上例的 as_of 應換成團隊實際凍結的 dataset version 時間。若在 CI 中每次都使用「現在」，你固定的只是執行時間，不是可追溯的資料版本。

這一步在驗證什麼：這一步直接對應④段落「dataset 不是一次性資產」的主張，但用具體 API 呼叫把它變成可執行的流程。五道審查問題（1-5）的順序不是隨便排的：先問有沒有個資和機密（能不能收），再問遮罩後還有沒有測試價值（收了有沒有用），最後才問治理（誰能改）。預期看到什麼：create_examples() 執行完，LangSmith UI 的 dataset 頁面會多出對應筆數，每筆的 metadata 欄位都能在 UI 上篩選。容易踩的坑：update_dataset_tag() 的 as_of 如果每次都傳當下時間，CI pipeline 拿到的其實是「執行當下的最新版本」而不是一個固定的基準，這代表兩次 release 之間即使程式碼沒變，光是 dataset 案例被別人默默加了一筆，評估結果也會跟著變化——這正是⑦段落 dataset_version_not_pinned 這個 block_when 條件想攔住的情境。

下一篇（下）接著跑 Step 9 到 Step 16：從 deterministic evaluator 比較 baseline 與 candidate、LLM-as-judge 的校準防線、feedback triage、investigation runbook，一路到 release gate 與本文結論。

這篇是 Learning SRE for the AI Era 系列的一部分：從 SRE Lab 到 Production AI Reliability。

Build → Trace → Break → Measure → Evaluate → Recover → Improve.